package com.sunlands.deskmate.message.handler.storager.unsync;

import com.alibaba.fastjson.JSONObject;
import com.sunlands.deskmate.exception.MessageSenderException;
import com.sunlands.deskmate.exception.SendExceptionUtil;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.message.handler.storager.MultipleSortedStorager;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;
import static com.sunlands.deskmate.config.RedisConfig.PREFIX;


/**
 * @author anjunling@sunlands.com
 * @author liude
 * <p>
 * unsync就是说消息已经发送, 还没有收到回执, 为了保证顺序, 在收到回执前不发送下一条
 */
@SuppressWarnings({"SpringJavaAutowiredFieldsWarningInspection", "WeakerAccess"})
@Slf4j
@Component("unsyncMessageStorager")
public class UnsyncMessageStorager {

    private static final String CAS = "desk_mate_deliver:cas:cas_";

    @Resource(name = "redisZSetStorager")
    private MultipleSortedStorager<Messages> storager;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;


    @Autowired
    private RedissonClient redissonClient;

    /**
     * 标记未同步消息最小ID
     */
    public static final String MESSAGE_UNSYNC = PREFIX + ":unsync:message_unsync_";

    /**
     * 标记已同步消息最大ID
     */
    public static final String MESSAGE_SYNCFLAG = PREFIX + ":sync:message_sync_flag_";

    /**
     * 每秒从未读消息库中读取的消息数量
     * <p>
     * private static final int userBatchCount = 500;
     * private static final int salesBatchCount = 10;
     */


    public long countUnreadMessage(String sessionKey) {
        return storager.count(unsyncKey(sessionKey));
    }

    /**
     * 所有未同步键
     *
     * @return List<String>
     */
    public List<String> unsyncKeys() {
        RedisCallback<Set<String>> redisCallback = redisConnection -> {
            Set<String> set = new ConcurrentSkipListSet<>();
            Cursor<byte[]> cursor = redisConnection.scan(new ScanOptions.ScanOptionsBuilder().match(MESSAGE_UNSYNC + "*").count(10000).build());
            while (cursor.hasNext()) {
                set.add(new String(cursor.next()));
            }
            return set;
        };
        return new ArrayList<>(redisTemplate.execute(redisCallback));
    }

    /**
     * 未同步消息列表
     *
     * @param sessionKey sessionKey
     * @return List<String> json格式的消息, 按照messageId排序
     */
    public Messages getUnreadMessages(String sessionKey) {
        List<Messages> messages = storager.rangeByFactorByCount(unsyncKey(sessionKey), 1);
        log.debug("batchMessageList={}", messages);
        return messages.isEmpty() ? null : messages.get(0);
    }

    public List<Messages> getUnreadMessages(String sessionKey, Long min, Long max) {
        List<Messages> messages = storager.rangeByFactor(unsyncKey(sessionKey), min, max);
        log.debug("batchMessageList={}", messages);
        return messages;
    }

    public boolean add(String sessionKey, Messages message, long messageId) {
        log.debug("storage message = {}", message);
        String key = unsyncKey(sessionKey);
        return storager.add(key, message, messageId);
    }

    public void del(String id) {
        String key = unsyncKey(id);
        storager.del(key);
        log.debug("delete key={}", key);
    }

    public void del(String id, long messageId) {
        String key = unsyncKey(id);
        log.info("delete key={} messageId = {}", key, messageId);
        storager.removeByFactor(key, messageId, messageId);
    }

    @Autowired
    private SendExceptionUtil sendExceptionUtil;

    /**
     * 登记最新已读消息ID
     * 前提是: messageId根据时间单调增长
     * <p>
     * 清除比messageId小的未读消息
     * 设置当前messageId为(最大)已读消息Id
     *
     * @param sessionKey sessionKey
     * @param messageId  messageId
     */
    public void setLatestReadMessageId(String sessionKey, long messageId) {

        log.debug("setLatestReadMessageId, sessionKey={},messageId={}", sessionKey, messageId);

        Long syncMsgId = getSyncFlagValue(sessionKey);


        //删除已同步消息
        String unsyncKey = unsyncKey(sessionKey);


        if (syncMsgId != null && syncMsgId > messageId) {
            Messages unreadMessages = getUnreadMessages(sessionKey);
            if (Objects.isNull(unreadMessages)) {
                return;
            }
            log.error("消息顺序异常： syncMsgId = {} messageId = {} unreadMessages = {}", syncMsgId, messageId, unreadMessages);
            sendExceptionUtil.sendException("消息顺序异常： syncMsgId = " + syncMsgId + " messageId = " + messageId + "unreadMessages = " + JSONObject.toJSON(unreadMessages), new MessageSenderException());
            storager.removeByFactor(unsyncKey, 0L, messageId);
            return;
        }
        //删除messageId之前的所有id
        //当messageId顺序不准确的时候, 就会造成多删和消息丢失
        storager.removeByFactor(unsyncKey, 0L, messageId);

        //记录已同步标记位
        String syncFlagKey = syncFlagKey(sessionKey);
        String newValue = messageId + "";
        String expectValue = syncMsgId == null ? null : syncMsgId.toString();

        RBucket<String> bucket = redissonClient.getBucket(casKey(syncFlagKey));
        boolean result;
        result = bucket.compareAndSet(expectValue, newValue);
        bucket.expire(10, TimeUnit.MINUTES);
        if (!result) {
            setLatestReadMessageId(sessionKey, messageId);
        }
        log.debug("setLatestReadMessageId,sessionKey={}, latestMessageId={}, previousMessageId={}, result={}", sessionKey, messageId, syncMsgId, result);
    }

    public static String unsyncKey(String id) {
        return MESSAGE_UNSYNC + id;
    }

    public static String syncFlagKey(String key) {
        return MESSAGE_SYNCFLAG + key;
    }


    /**
     * 获取已同步消息标识位值
     *
     * @param sessionKey sessionKey
     * @return Long 已同步消息的最大ID
     */
    public Long getSyncFlagValue(String sessionKey) {
        String key = casKey(syncFlagKey(sessionKey));
        //已同步标识位value
        RBucket<String> bucket = redissonClient.getBucket(key);
        String value = bucket.get();
        log.debug("getSyncFlagValue, key={}, value={}", key, value);
        return Objects.isNull(value) ? null : Long.parseLong(value);
    }


    public static String casKey(String key) {
        return CAS + key;
    }
}
